package com.tc.vms.kafka;

import com.tc.vms.MessageWrap;
import com.tc.vms.log.VmsClientLogger;
import org.slf4j.Logger;

import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by yonghua.zhang on 2016/5/16.
 */
public class ProcessQueue {
    private final Logger log = VmsClientLogger.getLog();
    public final static long RebalanceLockMaxLiveTime = 30000;
    public final static long RebalanceLockInterval = 20000;
    private final static long PullMaxIdleTime = 120000;

    private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
    private final ConcurrentSkipListMap<Long, MessageWrap> msgTreeMap = new ConcurrentSkipListMap<Long, MessageWrap>();
    private volatile long queueOffsetMax = 0L;
    private final AtomicLong msgCount = new AtomicLong();

    private volatile boolean dropped = false;
    private volatile long lastPullTimestamp = System.currentTimeMillis();
    private volatile long lastConsumeTimestamp = System.currentTimeMillis();
    private volatile long lastLockTimestamp = System.currentTimeMillis();

    private final Lock lockConsume = new ReentrantLock();

    private volatile boolean locked = false;

    private volatile boolean consuming = false;
    /*private final TreeMap<Long, MessageExt> msgTreeMapTemp = new TreeMap<Long, MessageExt>();*/
    private final AtomicLong tryUnlockTimes = new AtomicLong(0);

    private volatile long msgAccCnt = 0;
}
